package org.example;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;

import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;

import java.io.IOException;

public class HyperbaseTest {

  private static final String ZK_QUORUM = "hbase.zookeeper.quorum";
  private static final String ZK_CLIENT_PORT = "hbase.zookeeper.property.clientPort";
  private static final String ZK_POS = "tdh03:2181,tdh04:2181,argo01:2181";
  private static final String ZK_PORT_VALUE = "2181";

  private static Connection conn;
  private static Configuration conf;

  public static void main(String[] args) throws IOException {
    initialize();
    scanData("test_zw:table_zw");
    closeConnection();
    System.exit(0);
  }

  private static void initialize() throws IOException {
    System.setProperty("java.security.krb5.conf", "./config/krb5.conf");
    System.setProperty("java.security.auth.login.config", "./config/jaas.conf");

    conf = HBaseConfiguration.create();
    conf.addResource(new Path("./config/hbase-site.xml"));
    conf.addResource(new Path("./config/hdfs-site.xml"));
    conf.addResource(new Path("./config/core-site.xml"));

    conf.set(ZK_QUORUM, ZK_POS);
    conf.set(ZK_CLIENT_PORT, ZK_PORT_VALUE);
    conf.set("hadoop.security.authentication", "kerberos");

    UserGroupInformation.setConfiguration(conf);
    UserGroupInformation.loginUserFromKeytab("hbase/tdh01@TDH", "./config/hyperbase.keytab");

    conn = ConnectionFactory.createConnection(conf);
  }

  private static void scanData(String tableName) throws IOException {
    try (Table table = conn.getTable(TableName.valueOf(tableName));
         ResultScanner scanner = table.getScanner(new Scan())) {

      for (Result result : scanner) {
        for (Cell cell : result.rawCells()) {
          String row = Bytes.toString(CellUtil.cloneRow(cell));
          String family = Bytes.toString(CellUtil.cloneFamily(cell));
          String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
          String value = Bytes.toString(CellUtil.cloneValue(cell));

          System.out.println("row: " + row + " family: " + family + " qualifier: " + qualifier + " value: " + value);
        }
        System.out.println("----------------------");
      }
    }
  }

  private static void closeConnection() {
    if (conn != null) {
      try {
        conn.close();
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  }
}
